package com.wunong.smart.bigdata.flink.job;

import com.wunong.smart.bigdata.flink.data.TimestampData;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * Interval join
 */
public class IntervalJoinJob {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment()
                .setParallelism(1);

        DataStream<TimestampData> stream1 = env.fromElements(
                TimestampData.create(1),
                TimestampData.create(2),
                TimestampData.create(3));

        // 时间便宜
        final int offsetMinutes = 1;

        long mills = System.currentTimeMillis();
        DataStream<TimestampData> stream2 = env.fromElements(
                // 在范围内
                TimestampData.create(1, 10),
                // 边界也会join
                TimestampData.create(2, 20, mills - offsetMinutes * 60 * 1000L),
                // 不在范围内不会join
                TimestampData.create(3, 30, mills - 2 * 60 * 1000L),
                // 没有关联数据不会join
                TimestampData.create(4, 20)
        );

        // 时间水印
        WatermarkStrategy<TimestampData> watermarkStrategy = getWatermarkStrategy();

        // 指定时间范围内合并
        SingleOutputStreamOperator<String> process = stream1.assignTimestampsAndWatermarks(watermarkStrategy)
                .keyBy(m -> String.valueOf(m.getId()))
                .intervalJoin(stream2.assignTimestampsAndWatermarks(watermarkStrategy).keyBy(v -> String.valueOf(v.getId())))
                // 时间区间在2分钟内
                .between(Time.minutes(-offsetMinutes), Time.minutes(offsetMinutes))
                // 指定处理器
                .process(new ProcessJoinFunction<TimestampData, TimestampData, String>() {
                    @Override
                    public void processElement(TimestampData left, TimestampData right,
                                               ProcessJoinFunction<TimestampData, TimestampData, String>.Context ctx,
                                               Collector<String> out) throws Exception {
                        out.collect(left.getId() + ":" + left.getData() + ":" + right.getData());
                    }
                });
        process.print("Result");

        env.execute(IntervalJoinJob.class.getName());
    }

    /**
     * 时间水印
     *
     * @return
     */
    public static WatermarkStrategy<TimestampData> getWatermarkStrategy() {
        // 时间水印，对事件时间内保证有序
        final Duration widowDuration = Duration.ofSeconds(300 * 3 / 5);
        WatermarkStrategy<TimestampData> watermarkStrategy = WatermarkStrategy
                .<TimestampData>forBoundedOutOfOrderness(widowDuration)
                .withTimestampAssigner(((message, timestamp) -> message.getTimestamp()));
        return watermarkStrategy;
    }

}